package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisWorker;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

import static com.hmdp.utils.RedisConstants.*;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Autowired
    private ISeckillVoucherService iSeckillVoucherService;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private RedisWorker redisWorker;

    /**
     * 建议放到yml配置文件中
     */
    private static final String LUA_FILE_PATH="seckill.lua";

    private static final String ORDER = "order";

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

     static {
         SECKILL_SCRIPT=new DefaultRedisScript<>();
         SECKILL_SCRIPT.setLocation(new ClassPathResource(LUA_FILE_PATH));
         SECKILL_SCRIPT.setResultType(Long.class);
     }

    /**
     * 异步线程
     */
    private static final ExecutorService SECKILL_ORDER_EXECUTOR= Executors.newSingleThreadExecutor();

    @PostConstruct
     public void init(){
        //如果指定stream队列关联的消费者组已经存在,则不进行处理
        if (!targetGroupExistsInStream(STREAM_QUEUE_NAME,STREAM_GROUP_NAME)) {
         //这里createGroup的mkStream为true,表示在创建消费者组时，如果关联的stream队列不存在,也会自动创建
        stringRedisTemplate.opsForStream().createGroup(STREAM_QUEUE_NAME,ReadOffset.from("0"),STREAM_GROUP_NAME);
        }
        //异步监听任务执行
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
     }

    private boolean targetGroupExistsInStream(String streamName,String groupName) {
        //如果stream队列不存在
        if(stringRedisTemplate.countExistingKeys(Collections.singletonList(streamName))==0){
         return false;
        }
        StreamInfo.XInfoGroups order_stream_groups = stringRedisTemplate.opsForStream().groups(streamName);
        Iterator<StreamInfo.XInfoGroup> iterator = order_stream_groups.iterator();
        while(iterator.hasNext()){
            StreamInfo.XInfoGroup xInfoGroup = iterator.next();
            if(xInfoGroup.groupName().equals(groupName)){
                return true;
            }
        }
        return false;
    }

    public class VoucherOrderHandler implements Runnable{
        @Override
        public void run() {
          while(true){
              try {
                  //1.获取队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >
                  List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                          //消费组的名字和消费者的名字
                          Consumer.from(STREAM_GROUP_NAME, STREAM_GROUP_CONSUMER_NAME),
                          StreamReadOptions.empty().count(1)
                                  //我这里设置为一直阻塞,直到有消息可读为止
                                  .block(Duration.ofSeconds(0)),
                          //从哪个Stream队列进行消息读取,此处读取方式为>
                          StreamOffset.create(STREAM_QUEUE_NAME, ReadOffset.lastConsumed())
                  );
                  if (handleMsgFromStream(list)) {
                      continue;
                  }
              } catch (Exception e) {
                  log.error("处理消息异常");
                  //处理Pending队列中消息
                  handlePendingList();
              }
          }
        }

        private void handlePendingList() {
            while(true){
                try {
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from(STREAM_GROUP_NAME, STREAM_GROUP_CONSUMER_NAME),
                            StreamReadOptions.empty().count(1),
                            //从头开始消费pending队列中所有消息
                            StreamOffset.create(STREAM_QUEUE_NAME, ReadOffset.from("0"))
                    );
                    if(handleMsgFromStream(list)){
                        //Pending队列中没有错误消息,那么直接退出循环
                        break;
                    }
                } catch (Exception e) {
                    //处理Pending队列中的异常消息
                    //可以在这里做一些异常记录等
                    try {
                        //避免循环频率过高
                        Thread.sleep(3000);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        }


        private boolean handleMsgFromStream(List<MapRecord<String, Object, Object>> list) {
            //2.判断订单信息是否为空
            if(list ==null || list.isEmpty()){
                //如果为null,说明没有消息
                return true;
            }
            //解析消息
            MapRecord<String, Object, Object> record = list.get(0);
            Map<Object, Object> value = record.getValue();
            VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
            //3.创建订单
            createVoucherOrder(voucherOrder);
            //4.确认消息
            stringRedisTemplate.opsForStream().acknowledge(STREAM_QUEUE_NAME,STREAM_GROUP_NAME,record.getId());
            return false;
        }

        /**
         * 保守起见，还会再次进行判断
         */
        private void createVoucherOrder(VoucherOrder voucherOrder) {
            Long userId = voucherOrder.getUserId();
            Long voucherId = voucherOrder.getVoucherId();
            //创建锁对象
            RLock lock = redissonClient.getLock(LOCK_ORDER_KEY + userId);
            //尝试获取分布式锁
            // 第一个参数为获取锁的最大等待时间(期间会重试)--默认-1,,失败直接返回
            //锁自动释放时间--默认30秒
            //时间单位
            boolean tryLock = lock.tryLock();
            if(!tryLock){
                log.error("用户["+userId+"]对"+voucherId+"优惠卷重复抢购");
                return;
            }
            //我们只需要确保下面这两行代码的集群并发问题被解决
            try{
                Integer count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();

                if(count>0){
                    log.error("用户["+userId+"]对"+voucherId+"优惠卷重复抢购");
                    return;
                }
            }finally {
                lock.unlock();
            }

            //6.扣减库存
            boolean success = iSeckillVoucherService.update()
                    .setSql("stock = stock -1")
                    .eq("voucher_id", voucherId)
                    .gt("stock",0)
                    .update();
            if(!success){
               log.error("库存扣减失败");
               return;
            }
            save(voucherOrder);
        }
    }

    @Override
    public Result seckillVoucher(Long voucherId) {
        Long uid = UserHolder.getUser().getId();
        Long orderId = redisWorker.nextId(ORDER);
        //1.执行lua脚本
        Long res = stringRedisTemplate.execute(SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(), uid.toString(),orderId.toString());
        //2.判断结果是否为0
        int r=res.intValue();
        if(r!=0){
            //说明redis中没有当前优惠卷的库存记录
            //去数据库查询该优惠卷是否存在,如果不存在,说明是恶意访问
            //如果存在,就更新redis记录
            if(r==3){
                if(handleUnKnownVoucherId(voucherId)){
                    seckillVoucher(voucherId);
                }else {
                    return Result.fail("指定优惠卷不存在");
                }
            }
            return Result.fail(r==1?"库存不足":"不能重复下单");
        }
        //3.为0,有购买资格,把下单信息保存到阻塞队列
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(uid);
        voucherOrder.setVoucherId(voucherId);
        //4.返回订单id
        return Result.ok(orderId);
    }

    private Boolean handleUnKnownVoucherId(Long voucherId) {
        SeckillVoucher voucher = iSeckillVoucherService.getById(voucherId);
        if(voucher==null){
            return false;
        }
        //更新redis记录
        stringRedisTemplate.opsForValue().set(SECKILL_STOCK_KEY+voucherId,voucher.getStock().toString());
       return true;
    }
}
